#define _GNU_SOURCE             /* See feature_test_macros(7) */
#include <arpa/inet.h>
#include <asm-generic/unistd.h>
#include <assert.h>
#include <endian.h>
#include <errno.h>
#include <execinfo.h>
#include <fcntl.h>
#include <ifaddrs.h>
#include <inttypes.h>
#include <limits.h>
#include <linux/ethtool.h>
#include <linux/reboot.h>
#include <linux/sockios.h>
#include <net/if.h>
#include <netinet/in.h>
#include <pthread.h>
#include <rte_atomic.h>
#include <rte_branch_prediction.h>
#include <rte_common.h>
#include <rte_cycles.h>
#include <rte_debug.h>
#include <rte_eal.h>
#include <rte_ethdev.h>
#include <rte_ether.h>
#include <rte_interrupts.h>
#include <rte_launch.h>
#include <rte_lcore.h>
#include <rte_log.h>
#include <rte_malloc.h>
#include <rte_mbuf.h>
#include <rte_memory.h>
#include <rte_mempool.h>
#include <rte_memzone.h>
#include <rte_pci.h>
#include <rte_per_lcore.h>
#include <rte_ring.h>
#include <rte_string_fns.h>
#include <sched.h>
#include <signal.h>
#include <stdio.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/ioctl.h>
#include <sys/reboot.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/syscall.h>
#include <sys/time.h>
#include <sys/types.h>
#include <syslog.h>
#include <time.h>
#include <unistd.h>
#include <zlib.h>
#include "config.h"
#include "list.h"
#include "misc.h"
#include "log.h"
#include "build_packet.h"

#define CT_MAX_TIMEOUT                      16
#define CT_TIMEOUT_MASK                     (CT_MAX_TIMEOUT - 1)
#define DEBUG_LEVEL_BLOCK_LOG               0x01
#define GET_IPH(m)                          ((struct iphdr* )(((char *)(m)->buf_addr) + L3_OFFSET(m)))
#define GET_IP6H(m)                         ((struct ip6_hdr* )(((char *)(m)->buf_addr) + L3_OFFSET(m)))
#define GET_NEW_TCP_TIMEOUT_HEAD(pid)       ({ unsigned i = (g_time + CT_MAX_TIMEOUT - 1) & CT_TIMEOUT_MASK; &ct_timeout_list_tables[pid][i]; })
#define GET_TCPH(m)                         ((struct tcphdr*)(((char *)(m)->buf_addr) + L4_OFFSET(m)))
#define GET_TIMEOUT_LIST_HEAD(pid)          ({ unsigned i = g_time & CT_TIMEOUT_MASK; &ct_timeout_list_tables[pid][i]; })
#define GET_UDPH(m)                         ((struct udphdr*)(((char *)(m)->buf_addr) + L4_OFFSET(m)))
#define LISTEN_ADDR                         "127.0.0.1"
#define LISTEN_PORT                         28991
#define MAC_LIST_SIZE                       32
#define MAX_CHANNEL                         16
#define MAX_PKT_BURST                       32
#define PACKET_BUF_SIZE                     (1024-1)
#define RCC_FILE                            "/mnt/storage/rcc"
#define SERVICE_TYPE_HTTP                   1
#define SERVICE_TYPE_HTTPS                  2
#define SERVICE_TYPE_SSL                    3
#define tcp_ack_seq_h(m)                    ntohl(GET_TCPH(m)->ack_seq)
#define tcp_seq_h(m)                        ntohl(GET_TCPH(m)->seq)
#define FLAG_LOCAL_CLIENT_CLOSE_OK     0x01
#define FLAG_LOCAL_CLIENT_HANDSHAKE_OK 0x02
#define FLAG_LOCAL_SERVER_CLOSE_OK     0x04
#define FLAG_LOCAL_SERVER_HANDSHAKE_OK 0x08
#define FLAG_NOTIFY_CLIENT_CLOSE       0x10
#define FLAG_NOTIFY_SERVER_CLOSE       0x20
#define LOCAL_CLIENT_SSL_INITED        0x40
#define LOCAL_SERVER_SSL_INITED        0x80

struct port_info {
    int reset;
    unsigned char reset_mac[6];
    uint64_t dropped;
    uint64_t fake_tx;
    uint64_t fake_tx_fail;
    uint64_t last_5s_rx;
    uint64_t last_5s_rx_byte;
    uint64_t last_rx;
    uint64_t last_rx_byte;
    uint64_t rx;
    uint64_t rx_byte;
    uint64_t tx;
    uint64_t tx_byte;
} __rte_cache_aligned;

struct packet_process_thread_info {
    int index;
    struct rte_eth_dev_tx_buffer *tx_buffer[MAX_PORT][MAX_QUEUE];
    uint64_t create_tcp_ct_count;
    uint64_t create_udp_ct_count;
    uint64_t destroy_tcp_ct_count;
    uint64_t destroy_udp_ct_count;
    uint64_t fake_tx;
    uint64_t rx;
    uint64_t rx_byte;
    uint64_t tx;
    uint64_t tx_byte;
    unsigned ct_count;
};

FILE *BLOCK_LOG_FP = NULL;
FILE *LLOG_FP = NULL;
char reset_device_name[MAX_CHANNEL][64];
const char *config_file_name = "/mnt/storage/http/web/data/mim.conf";
extern int save_bingdu_file_switch;
long image_bloom_filter_life = 7*24*60*60;
int broadcast_port = -1;
int bypass = 0;
int capture_count = 0;
int capture_count_now = 0;
int channel_count = 1;
int channel_port_reset[MAX_CHANNEL];
int debug_level = 0;
int dump_http_host_errno = 0;
int err_page_len = 0;
int g_mysql_connect_ok = 0;
int log_mem_alloc = 1;
int print_stats_interval = 0;
int reset_port_count = 0;
int reset_ports[16];
int save_packet_fd = -1;
int super_filter_mode = 0;
pthread_mutex_t log_file_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t save_packet_lock = PTHREAD_MUTEX_INITIALIZER;
rte_atomic32_t queue_index = RTE_ATOMIC32_INIT(0);
struct if_stats *if_stats = NULL;
struct ipv6_pool *ipv6_pool = NULL;
struct my_hlist_head *trackers[MAX_QUEUE];
struct my_list_head *ct_timeout_list_tables[MAX_QUEUE] = {};
struct my_list_head tuple_pool_lists [MAX_QUEUE] = {};
struct packet_process_thread_info packet_process_thread_infos[MAX_QUEUE] = {};
struct port_info port_infos[RTE_MAX_ETHPORTS] = {};
struct rte_mempool *mp_for_fake_packet = NULL;
struct rte_mempool *mp_for_tcp_sort = NULL;
time_t capture_start_time;
time_t capture_stop_time;
time_t g_time;
time_t last_check_status_time = 0;
uint32_t ct_htable_size = 1;
uint32_t tcp_sort_cache_max = 8;
uint64_t g_bps = 0;
uint64_t g_pps = 0;
uint64_t vpn_producter_out_ok_count = 0;
uint64_t match_nofilter_count = 0;
uint64_t producter_filter_count = 0;
uint64_t uploader_filter_count = 0;
unsigned ct_htable_bit = 10;
unsigned ct_htable_mask;
unsigned short dbport = 28096;
volatile int cid_list_count;
volatile uint64_t cid_list[64];
int fake_pkg_use = 0;

int init_packinf(struct rte_mbuf *m, unsigned queue);

void    l2fwd_simple_forward(struct rte_mbuf *m, unsigned portid);
#define l2fwd_simple_forward(a, b) ({ init_packinf(a, 0); \
        if (0 && ntohs(GET_TCPH(a)->source) == 443) LOGERR("SSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSS\n"); \
        LOGERR(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> port %d %u --> %u seq %u ack_seq %u ack %d win %u app_data_len %d\n"\
                , !b \
                , ntohs(GET_TCPH(a)->source) \
                , ntohs(GET_TCPH(a)->dest) \
                , ntohl(GET_TCPH(a)->seq), ntohl(GET_TCPH(a)->ack_seq)\
                , GET_TCPH(a)->ack\
                , ntohs(GET_TCPH(a)->window)\
                , APP_DATA_LEN(a)); l2fwd_simple_forward(a , b); })

static inline void add_packet_to_sended_list(tuple_t *tuple, struct rte_mbuf *m, int is_local_server)
{
    GET_PACKINF(m)->sended_packet_next = NULL;
    SSL_CONTEXT *context = tuple->ssl_context;
    context->sended_unack_len[is_local_server] += GET_PACKINF(m)->app_data_len;
    if (context->sended_packet_last[is_local_server]) {
        GET_PACKINF(context->sended_packet_last[is_local_server])->sended_packet_next = m;
        context->sended_packet_last[is_local_server] = m;
    } else {
        context->sended_packets[is_local_server] = m;
        context->sended_packet_last[is_local_server] = m;
    }
    context->sended_packet_count[is_local_server]++;
}

static inline void send_cache_data(tuple_t *tuple, int is_local_server)
{
    SSL_CONTEXT *context = tuple->ssl_context;
    while (context->unsend_packets[is_local_server]) {
        if (context->sended_unack_len[is_local_server] >= 16413 || context->sended_packet_count[is_local_server] >= 16) {
            LOGERR("context->sended_unack_len >= 16413 || context->sended_packet_count >= 16 %d %d\n", context->sended_unack_len[is_local_server], context->sended_packet_count[is_local_server]);
            break;
        }
        struct rte_mbuf *m = context->unsend_packets[is_local_server];
        rte_mbuf_refcnt_update(m, 1);
        l2fwd_simple_forward(m, !GET_PACKINF(m)->send_to_client_port);
        context->unsend_packets[is_local_server] = GET_PACKINF(m)->unsend_packet_next;
        context->unsend_len[is_local_server] -= GET_PACKINF(m)->app_data_len;
        if (context->unsend_packets[is_local_server] == NULL) context->unsend_packet_last[is_local_server] = NULL;
        add_packet_to_sended_list(tuple, m, 1);
        LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n",
                fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
        context->unsend_packet_count[is_local_server]--;
        //debug
        assert(context->unsend_packet_count[is_local_server] >= 0);
        assert(fake_pkg_use == context->unsend_packet_count[1] + context->unsend_packet_count[0] + context->sended_packet_count[1] + context->sended_packet_count[0]);
    }
}
#define send_cache_data_to_client(a, b) ({ LOGERR("calling send_cache_data_to_client\n"); send_cache_data_to_client(a, b); })

static inline void resend_data(tuple_t *tuple, int is_local_server)
{
    if (tuple) {
        struct rte_mbuf *m;
        int n = 0;
        for (m = tuple->ssl_context->sended_packets[is_local_server]; m; m = GET_PACKINF(m)->sended_packet_next) {
            rte_mbuf_refcnt_update(m, 1);
            l2fwd_simple_forward(m, !GET_PACKINF(m)->send_to_client_port);
            if (++n >= 1) return;
        }
    }
}

uint32_t tuple_pool_size = 100000;
uint64_t ignore_ct_count = 0;
uint64_t ignore_ct_server_access_count = 0;

static inline int send_to(tuple_t *tuple, const uint8_t *data, size_t data_len, int ack, int fin, int rst, int is_local_server)
{
    SSL_CONTEXT *context = tuple->ssl_context;
    if (context->sended_packet_count[is_local_server] + context->unsend_packet_count[is_local_server] >= 32) { LOGERR("context->sended_packet_count + context->unsend_packet_count >= 32\n"); return -1; }
    int packlen;
    unsigned char fake_packet[2048];
    uint32_t seq = tuple->to_client_my_seq;
    int win = sizeof(context->recv_buf[is_local_server]) - context->recv_len[is_local_server] + sizeof(context->send_buf[!is_local_server]) - context->send_len[!is_local_server];
    if (win < 0) win = 0;
    if (is_local_server) {
        //LOGERR("to client, set seq tuple->to_client_my_seq %u ack_seq context->last_deal_pkg_seq_add_len[1] %u\n", tuple->to_client_my_seq, context->last_deal_pkg_seq_add_len[1]);
        packlen = build_tcp_packet(fake_packet,
                tuple->dstmac, tuple->srcmac,
                tuple->dstip[3], tuple->srcip[3],
                tuple->dstport, tuple->srcport,
                htonl(tuple->to_client_my_seq), htonl(context->last_deal_pkg_seq_add_len[is_local_server]),
                0, ack, fin, rst, win, data, data_len, "", 0);
        tuple->to_client_my_seq += data_len + fin;
    } else {
        //LOGERR("to server, set seq tuple->to_client_my_seq %u ack_seq context->last_deal_pkg_seq_add_len[0] %u\n", tuple->to_server_my_seq, context->last_deal_pkg_seq_add_len[0]);
        packlen = build_tcp_packet(fake_packet,
                tuple->srcmac, tuple->dstmac,
                tuple->srcip[3], tuple->dstip[3],
                tuple->srcport, tuple->dstport,
                htonl(tuple->to_server_my_seq), htonl(context->last_deal_pkg_seq_add_len[is_local_server]),
                0, ack, fin, rst, win, data, data_len, "", 0);
        tuple->to_server_my_seq += data_len + fin;
    }

    struct rte_mbuf *f = NULL;
    if (fake_pkg_use < FAKE_PKG_POOL_SIZE) f = rte_pktmbuf_alloc(mp_for_fake_packet);
    if (f == NULL) {
        LOGERR("fail to rte_pktmbuf_alloc\n");
        LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n",
                fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
        return MBEDTLS_ERR_SSL_ALLOC_FAILED;
    }
    fake_pkg_use++;
    memcpy(rte_pktmbuf_append(f, packlen), fake_packet, packlen);
    if (data_len > 0) {
        GET_PACKINF(f)->seq = seq;
        GET_PACKINF(f)->app_data_len = data_len;
        GET_PACKINF(f)->send_to_client_port = tuple->client_port;
        if (context->sended_unack_len[is_local_server] >= 16413) {
            GET_PACKINF(f)->unsend_packet_next = NULL;
            if (context->unsend_packet_last[is_local_server]) {
                GET_PACKINF(context->unsend_packet_last[is_local_server])->unsend_packet_next = f;
                context->unsend_packet_last[is_local_server] = f;
            } else {
                context->unsend_packets[is_local_server] = f;
                context->unsend_packet_last[is_local_server] = f;
            }
            context->unsend_packet_count[is_local_server]++;
            context->unsend_len[is_local_server] += GET_PACKINF(f)->app_data_len;
        } else {
            rte_mbuf_refcnt_update(f, 1);
            l2fwd_simple_forward(f, is_local_server ? !tuple->client_port : !tuple->server_port);
            add_packet_to_sended_list(tuple, f, is_local_server);
        }
    } else {
        //LOGERR("just ack it\n");
        l2fwd_simple_forward(f, is_local_server ? !tuple->client_port : !tuple->server_port);
        fake_pkg_use--;
    }
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n",
            fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
    //debug
    assert(fake_pkg_use == context->unsend_packet_count[1] + context->unsend_packet_count[0] + context->sended_packet_count[1] + context->sended_packet_count[0]);
    return 0;
}
#define send_to(tuple, data, data_len, ack, fin, rst, is) ({ LOGERR("calling send_to app_data_len %lu %d\n", data_len, is); send_to(tuple, data, data_len, ack, fin, rst, is); })

static inline int bio_send_to_client(void *ctx, const unsigned char *buf, size_t len)
{
    tuple_t *tuple = ctx;
    size_t rest_len = len;
    do {
        size_t n = MIN(rest_len, 1460);
        if (send_to(tuple, buf, n, 1, 0, 0, 1)) break;
        buf += n;
        rest_len -= n;
    } while (rest_len);
    if (len == rest_len) {
        LOGERR("can not send now, please send later, return MBEDTLS_ERR_SSL_WANT_WRITE\n");
        return MBEDTLS_ERR_SSL_WANT_WRITE;
    }
    LOGERR("send %lu %lu\n", len , len - rest_len);
    return len - rest_len;
}

static inline int bio_send_to_server(void *ctx, const unsigned char *buf, size_t len)
{
    tuple_t *tuple = ctx;
    size_t rest_len = len;
    do {
        size_t n = MIN(rest_len, 1460);
        if (send_to(tuple, buf, n, 1, 0, 0, 0)) break;
        buf += n;
        rest_len -= n;
    } while (rest_len);
    if (len == rest_len) {
        LOGERR("can not send now, please send later, return MBEDTLS_ERR_SSL_WANT_WRITE\n");
        return MBEDTLS_ERR_SSL_WANT_WRITE;
    }
    LOGERR("send %lu %lu\n", len , len - rest_len);
    return len - rest_len;
}

static int bio_recv(void *ctx, unsigned char *buf, size_t len, int is_local_server)
{
    tuple_t *tuple = ctx;
    SSL_CONTEXT *context = tuple->ssl_context;
    if (context->recv_len[is_local_server] <= 0) {
#if 0
        if (tuple->flag & TUPLE_CT_FLAG_CLIENT_FIN) {
            //send_to(tuple, (const uint8_t *)"", 0, 1, 1, 0, 0, 0); 
            LOGERR("tuple %p see TUPLE_CT_FLAG_FIN, return MBEDTLS_ERR_NET_CONN_RESET\n", tuple);
            //return MBEDTLS_ERR_SSL_PEER_CLOSE_NOTIFY;
            return MBEDTLS_ERR_NET_CONN_RESET;
        }   
#endif
        LOGERR("tuple %p context->recv_from_client_len <= 0, return want read\n", tuple);
        return MBEDTLS_ERR_SSL_WANT_READ;
    }   
    //if (sizeof(context->send_to_server_buf) - context->unsend_to_server_len < 16413) return MBEDTLS_ERR_SSL_WANT_READ;
    size_t n = MIN(context->recv_len[is_local_server], len);
    memcpy(buf, context->recv_buf[is_local_server], n); 
    context->recv_len[is_local_server] -= n;
    memmove(context->recv_buf[is_local_server], context->recv_buf[is_local_server]+n, context->recv_len[is_local_server]);
    LOGERR("tuple %p recv %lu %lu left %lu\n", tuple, len, n, context->recv_len[is_local_server]); 
    return n;
}

static int bio_recv_from_client(void *ctx, unsigned char *buf, size_t len)
{
    return bio_recv(ctx, buf, len, 1);
}

static int bio_recv_from_server(void *ctx, unsigned char *buf, size_t len)
{
    return bio_recv(ctx, buf, len, 0);
}

SSL_CONTEXT ssll = {0};
int ssl_context_used = 0;
static inline SSL_CONTEXT *ssl_context_get(void)
{
    LOGERR("GGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGG\n");
    if (ssl_context_used) return NULL;
    ssl_context_used = 1;
    //debug
    memset(&ssll, 0, sizeof(ssll));
    return &ssll;
}

static inline void ssl_context_put(SSL_CONTEXT *context)
{
    LOGERR("PPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPP\n");
    struct rte_mbuf *m, *n;
    ssl_context_used = 0;
    if (context->flag & LOCAL_CLIENT_SSL_INITED) mbedtls_ssl_session_reset(&context->ssl[0]);
    if (context->flag & LOCAL_SERVER_SSL_INITED) mbedtls_ssl_session_reset(&context->ssl[1]);
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n", fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
    for (m = context->recved_packets[0]; m; m = n) { n = GET_PACKINF(m)->recv_packet_next; rte_pktmbuf_free(m); fake_pkg_use--; }
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n", fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
    for (m = context->recved_packets[1]; m; m = n) { n = GET_PACKINF(m)->recv_packet_next; rte_pktmbuf_free(m); fake_pkg_use--; }
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n", fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
    for (m = context->unsend_packets[0]; m; m = n) { n = GET_PACKINF(m)->unsend_packet_next; rte_pktmbuf_free(m); fake_pkg_use--; }
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n", fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
    for (m = context->unsend_packets[1]; m; m = n) { n = GET_PACKINF(m)->unsend_packet_next; rte_pktmbuf_free(m); fake_pkg_use--; }
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n", fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
    for (m = context->sended_packets[0]; m; m = n) { n = GET_PACKINF(m)->sended_packet_next; rte_pktmbuf_free(m); fake_pkg_use--; }
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n", fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
    for (m = context->sended_packets[1]; m; m = n) { n = GET_PACKINF(m)->sended_packet_next; rte_pktmbuf_free(m); fake_pkg_use--; }
    LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n", fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
}

static tuple_t *create_tuple(struct rte_mbuf *m, int queue, uint8_t protocol)
{
    struct tcphdr *tcph = GET_TCPH(m);
    uint16_t srcport = tcph->source;
    uint16_t dstport = tcph->dest;
    tuple_t *tuple = NULL;

    if (my_list_empty(&tuple_pool_lists[queue])) { ignore_ct_count++; return NULL; }
    tuple = my_list_first_entry(&tuple_pool_lists[queue], tuple_t, list_node);
    my_list_del(&tuple->list_node);
    tuple->protocol = protocol;
    tuple->ts = g_time;
    tuple->create_ts = g_time;
    if (protocol == IPPROTO_UDP) {
    } else if (protocol == IPPROTO_TCP) {
        tuple->server_port = m->port;
        tuple->client_port = !m->port;
        memcpy(tuple->srcmac, ((uint8_t *)m->buf_addr) + m->data_off, 6);
        memcpy(tuple->dstmac, ((uint8_t *)m->buf_addr) + m->data_off + 6, 6);
        if (IS_IPV6(m) == 0) {
            memset(tuple->srcip, 0, 12);
            memset(tuple->dstip, 0, 12);
            struct iphdr *iph  = GET_IPH(m);
            tuple->srcip[3] = iph->daddr;
            tuple->dstip[3] = iph->saddr;
        } else {
            struct ip6_hdr *ip6h = GET_IP6H(m);
            memcpy(tuple->srcip, &ip6h->ip6_dst, 16);
            memcpy(tuple->dstip, &ip6h->ip6_src, 16);
        }
        tuple->srcport = tcph->dest;
        tuple->dstport = tcph->source;
        if (protocol == IPPROTO_TCP) {
            tuple->to_server_my_seq = tuple->wait_seq[0] = ntohl(tcph->ack_seq);
            tuple->to_client_my_seq = tuple->wait_seq[1] = ntohl(tcph->seq) + 1; 
        }
        my_list_add(&tuple->list_node, GET_NEW_TCP_TIMEOUT_HEAD(queue));
    }

    struct my_hlist_head *head;
    unsigned hash = (ntohl(tuple->srcip[3]) + ntohl(tuple->dstip[3]) + ntohs(srcport) + ntohs(dstport)) & ct_htable_mask;
    head = &trackers[queue][hash];
    LOGMSG("create tuple at list %p\n", head);
    my_hlist_add_head(&tuple->node, head);
    packet_process_thread_infos[queue].ct_count++;
    LOGMSG("queue %d ct_count %d\n", queue, packet_process_thread_infos[queue].ct_count);
    switch (protocol) {
        case 6:
            packet_process_thread_infos[queue].create_tcp_ct_count++;
            break;
    }
    return tuple;
}

static void release_tuple(unsigned queue, tuple_t *tuple)
{
    struct rte_mbuf *next;
    packet_process_thread_infos[queue].ct_count--;
    if (tuple->tcp_sort_cache[0]) {
        struct rte_mbuf *curr = tuple->tcp_sort_cache[0];
        while (curr) {
            next = GET_PACKINF(curr)->tcp_sort_next;
            rte_pktmbuf_free(curr);
            curr = next;
        }
        tuple->tcp_sort_cache[0] = NULL;
    }
    if (tuple->tcp_sort_cache[1]) {
        struct rte_mbuf *curr = tuple->tcp_sort_cache[1];
        while (curr) {
            next = GET_PACKINF(curr)->tcp_sort_next;
            rte_pktmbuf_free(curr);
            curr = next;
        }
        tuple->tcp_sort_cache[1] = NULL;
    }
    if (tuple->ssl_context) { ssl_context_put(tuple->ssl_context); tuple->ssl_context = NULL; }
    tuple->flag = 0;
    tuple->packet_count = 0;
    tuple->service = 0;
    my_hlist_del(&tuple->node);
    my_list_del(&tuple->list_node);
    my_list_add(&tuple->list_node, &tuple_pool_lists[queue]);
}
//#define release_tuple(fmt, ...) ({ LOGERR("calling release_tuple\n"); release_tuple(fmt, ##__VA_ARGS__); })
//#define release_tuple(fmt, ...) ({ LOGERR("calling release_tuple\n"); release_tuple(fmt, ##__VA_ARGS__); exit(1); })

void check_tuple_timeout(unsigned queue);
void check_tuple_timeout(unsigned queue)
{
    tuple_t *tuple, *n;
    struct my_list_head *head = GET_TIMEOUT_LIST_HEAD(queue);
    my_list_for_each_entry_safe(tuple, n, head, list_node) { 
        release_tuple(queue, tuple);
    }
}

//return 0 ok
//return 1 cache
//return 2 full
//return 3 same
//return 4 fail alloc
//return 100 old
static inline int tcp_sort(tuple_t *tuple, struct rte_mbuf *m)
{
    uint8_t dir = PACK_DIR(m);
    uint32_t seq = tcp_seq_h(m);
    uint32_t wait_seq;

    if (seq == tuple->wait_seq[dir]) {
        wait_seq = tcp_seq_h(m) + APP_DATA_LEN(m) + GET_TCPH(m)->fin; tuple->wait_seq[dir] = wait_seq;
        struct rte_mbuf *cached = tuple->tcp_sort_cache[dir];
        while (cached) {
            struct rte_mbuf *cached_next = GET_PACKINF(cached)->tcp_sort_next;
            if (tcp_seq_h(cached) == wait_seq) {
                GET_PACKINF(cached)->tcp_sort_next = NULL;
                GET_PACKINF(m)->tcp_sort_next = cached;
                m = cached;
                wait_seq = tcp_seq_h(m) + APP_DATA_LEN(m) + GET_TCPH(m)->fin; tuple->wait_seq[dir] = wait_seq;
                tuple->tcp_sort_cache_count[dir]--;
            } else if (tcp_seq_h(cached) < wait_seq) {
                rte_pktmbuf_free(cached);
                tuple->tcp_sort_cache_count[dir]--;
            } else break;
            cached = cached_next;
        }
        tuple->tcp_sort_cache[dir] = cached;
        return 0;
    } else if (seq > tuple->wait_seq[dir]) {
        if (tuple->tcp_sort_cache_count[dir] >= tcp_sort_cache_max) return 2;
        if (tuple->tcp_sort_cache[dir] == NULL) {
            struct rte_mbuf *c = rte_pktmbuf_clone(m, mp_for_tcp_sort); if (c == NULL) { LOGFEW("fail to rte_pktmbuf_clone\n"); return 4; }
            memcpy(GET_PACKINF(c), GET_PACKINF(m), sizeof(struct packinf));
            tuple->tcp_sort_cache[dir] = c;
            tuple->tcp_sort_cache_count[dir] = 1;
            return 1;
        }
        if (seq < tcp_seq_h(tuple->tcp_sort_cache[dir])) {
            struct rte_mbuf *c = rte_pktmbuf_clone(m, mp_for_tcp_sort); if (c == NULL) { LOGFEW("fail to rte_pktmbuf_clone\n"); return 4; }
            memcpy(GET_PACKINF(c), GET_PACKINF(m), sizeof(struct packinf));
            GET_PACKINF(c)->tcp_sort_next = tuple->tcp_sort_cache[dir];
            tuple->tcp_sort_cache[dir] = c;
            tuple->tcp_sort_cache_count[dir]++;
            return 1;
        }
        if (seq == tcp_seq_h(tuple->tcp_sort_cache[dir])) return 3;
        struct rte_mbuf *tmp;
        for (tmp = tuple->tcp_sort_cache[dir]; GET_PACKINF(tmp)->tcp_sort_next; tmp = GET_PACKINF(tmp)->tcp_sort_next) {
            if (tcp_seq_h(GET_PACKINF(tmp)->tcp_sort_next) == seq) return 3;
            if (tcp_seq_h(GET_PACKINF(tmp)->tcp_sort_next) > seq) {
                struct rte_mbuf *c = rte_pktmbuf_clone(m, mp_for_tcp_sort); if (c == NULL) { LOGFEW("fail to rte_pktmbuf_clone\n"); return 4; }
                memcpy(GET_PACKINF(c), GET_PACKINF(m), sizeof(struct packinf));
                GET_PACKINF(c)->tcp_sort_next = GET_PACKINF(tmp)->tcp_sort_next;
                GET_PACKINF(tmp)->tcp_sort_next = c;
                tuple->tcp_sort_cache_count[dir]++;
                return 1;
            }
        }
        struct rte_mbuf *c = rte_pktmbuf_clone(m, mp_for_tcp_sort); if (c == NULL) { LOGFEW("fail to rte_pktmbuf_clone\n"); return 4; }
        memcpy(GET_PACKINF(c), GET_PACKINF(m), sizeof(struct packinf));
        GET_PACKINF(tmp)->tcp_sort_next = c;
        tuple->tcp_sort_cache_count[dir]++;
        return 1;
    }
    return 100;
}

static inline tuple_t *find_tuple(struct rte_mbuf *m, unsigned queue, uint8_t protocol)
{
    tuple_t *tuple = NULL;
    struct tcphdr *tcph = GET_TCPH(m);
    uint32_t srcip[4];
    uint32_t dstip[4];
    uint16_t srcport = tcph->source;
    uint16_t dstport = tcph->dest;
    if (IS_IPV6(m) == 0) {
        struct iphdr *iph = GET_IPH(m);
        memset(srcip, 0, 12);
        memset(dstip, 0, 12);
        srcip[3] = iph->saddr;
        dstip[3] = iph->daddr;
    } else {
        struct ip6_hdr *ip6h = GET_IP6H(m);
        memcpy(srcip, &ip6h->ip6_src, 16);
        memcpy(dstip, &ip6h->ip6_dst, 16);
    }

    unsigned hash = (ntohl(srcip[3]) + ntohl(dstip[3]) + ntohs(srcport) + ntohs(dstport)) & ct_htable_mask;
    struct my_hlist_head *head = &trackers[queue][hash];
    struct my_hlist_node *node, *n;
    LOGMSG("finding tuple at list %p\n", head);
    my_hlist_for_each_entry_safe(tuple, node, n, head, node) {
        if (tuple->srcport == srcport && tuple->dstport == dstport && tuple->protocol == protocol &&
                memcmp(tuple->srcip, srcip, 16) == 0 && memcmp(tuple->dstip, dstip, 16) == 0) {
            tuple->ts = g_time;
            PACK_DIR(m) = 0;
            break;
        }
        if (tuple->srcport == dstport && tuple->dstport == srcport && tuple->protocol == protocol &&
                memcmp(tuple->srcip, dstip, 16) == 0 && memcmp(tuple->dstip, srcip, 16) == 0) {
            tuple->ts = g_time;
            PACK_DIR(m) = 1;
            break;
        }
    }
    if (node) {
        my_list_del(&tuple->list_node);
        switch (protocol) {
            case 6 : my_list_add(&tuple->list_node, GET_NEW_TCP_TIMEOUT_HEAD(queue)); break;
        }
        LOGMSG("found\n");
        return tuple;
    }
    LOGMSG("not found\n");
    return NULL;
}


#define CHECK_RULE_CHECK_FILTER             0x01
#define CHECK_RULE_CHECK_FILTER_WITH_LOG    0x02
#define CHECK_RULE_CHECK_MONITOR            0x04
#define CHECK_RULE_CHECK_NOFILTER           0x08

int packet_process_thread_init(int packet_process_thread_index);
int packet_process_thread_init(int packet_process_thread_index)
{
    ct_htable_size = 1<<ct_htable_bit;
    ct_htable_mask = ct_htable_size - 1;
    uint32_t i;
    tuple_t *tuple;
    char name[64]; sprintf(name, "packet_ring_%d\n", packet_process_thread_index);
    LLOG("packet_process_thread %d tuple pool size %u mem %luB\n", packet_process_thread_index, tuple_pool_size, tuple_pool_size * sizeof(tuple_t));
    //tuple = (tuple_t *)rte_malloc("", tuple_pool_size * sizeof(tuple_t), 0); if (tuple == NULL) die("fail to ret_malloc\n");
    tuple = (tuple_t *)malloc(tuple_pool_size * sizeof(tuple_t));              if (tuple == NULL) die("fail to     malloc\n");
    memset(tuple, 0, tuple_pool_size * sizeof(tuple_t));
    JDIS_INIT_LIST_HEAD(&tuple_pool_lists[packet_process_thread_index]);
    for (i = 0; i < tuple_pool_size; i++) {
        my_list_add(&tuple->list_node, &tuple_pool_lists[packet_process_thread_index]);
        tuple++;
    }
    LLOG("packet_process_thread %d ct_htable size %u mem %luB\n", packet_process_thread_index, ct_htable_size, ct_htable_size*sizeof(struct my_hlist_head));
    trackers[packet_process_thread_index] = calloc(ct_htable_size, sizeof(struct my_hlist_head)); if (trackers[packet_process_thread_index] == NULL) die("fail to calloc[%m]\n");
    for (i = 0; i < ct_htable_size; i++) INIT_JDIS_HLIST_HEAD(&trackers[packet_process_thread_index][i]);
    ct_timeout_list_tables[packet_process_thread_index] = malloc(CT_MAX_TIMEOUT * sizeof(struct my_list_head));
    for (i = 0; i < CT_MAX_TIMEOUT; i++) JDIS_INIT_LIST_HEAD(&ct_timeout_list_tables[packet_process_thread_index][i]);
    return 0;
}

static inline char *ssl_err_str(int i)
{
    static char error_buf[100]= {};
    mbedtls_strerror(i, error_buf, 100);
    return error_buf;
}

static inline int handshack(tuple_t *tuple, int is_local_server)
{
    if ((tuple->flag & TUPLE_CT_FLAG_SSL_INIT) == 0) return -1;
    int ret = mbedtls_ssl_handshake(&tuple->ssl_context->ssl[is_local_server]);
    switch (ret) {
        case 0:
            LOGERR("***************************tuple %p local %s handshake server ok******************************\n", tuple, is_local_server ? "server" : "client");
            tuple->ssl_context->flag |= is_local_server ? FLAG_LOCAL_SERVER_HANDSHAKE_OK : FLAG_LOCAL_CLIENT_HANDSHAKE_OK;
            return 0;
        case MBEDTLS_ERR_SSL_WANT_READ:
            LOGERR("tuple %p local %s handshake server return MBEDTLS_ERR_SSL_WANT_READ\n", tuple, is_local_server ? "server" : "client");
            return 0;
        case MBEDTLS_ERR_SSL_WANT_WRITE:
            LOGERR("tuple %p local %s handshake server return MBEDTLS_ERR_SSL_WANT_WRITE\n", tuple, is_local_server ? "server" : "client");
            return 0;
        default:
            LOGERR("tuple %p local %s handshake server error %d[%s]\n", tuple, is_local_server ? "server" : "client", ret, ssl_err_str(ret));
    }
    return -1;
}

static inline int trans_data(tuple_t *tuple, SSL_CONTEXT *context, int is_local_server)
{
    if ((context->flag & FLAG_LOCAL_CLIENT_HANDSHAKE_OK) == 0 || (context->flag & FLAG_LOCAL_SERVER_HANDSHAKE_OK) == 0) return 0;
    if (tuple) {}
    LOGERR("is_local_server %d\n", is_local_server);
    LOGERR("%s --> %s %lu\n", is_local_server ? "client" : "server", is_local_server ? "server" : "client", context->send_len[is_local_server]);
    while (context->send_len[is_local_server] > 0) {
        int ret = mbedtls_ssl_write(&context->ssl[is_local_server], context->send_buf[is_local_server], context->send_len[is_local_server]);
        if (ret <= 0) {
            //if(ret == MBEDTLS_ERR_NET_CONN_RESET) return -1;
            //if(ret == MBEDTLS_ERR_NET_CONN_RESET) return -1;
            if (ret == MBEDTLS_ERR_SSL_WANT_WRITE) return MBEDTLS_ERR_SSL_WANT_WRITE;
            LOGERR("error, mbedtls_ssl_write return -0x%x\n", -ret);
            return ret;
        }
        //DUMP(context->send_buf[is_local_server], ret);
        int rest_len = context->send_len[is_local_server] - ret;
        memcpy(context->send_buf[is_local_server], context->send_buf[is_local_server] + ret, rest_len);
        context->send_len[is_local_server] = rest_len;
    }
    return 0;
}
#define trans_data(fmt, ...) ({ LOGERR("calling trans_data\n"); trans_data(fmt, ##__VA_ARGS__); })

static inline int deal_recved_data(tuple_t *tuple, int is_local_server)
{
    SSL_CONTEXT *context = tuple->ssl_context;
    for (;;) {
        size_t buf_len = sizeof(context->send_buf[!is_local_server]) - context->send_len[!is_local_server];
        uint8_t *buf   = context->send_buf[!is_local_server]         + context->send_len[!is_local_server];
        if (buf_len <= 0 || context->flag & FLAG_LOCAL_CLIENT_CLOSE_OK || context->flag & FLAG_NOTIFY_CLIENT_CLOSE) break;
        int ret = mbedtls_ssl_read(&context->ssl[is_local_server], buf, buf_len);
        if (ret == MBEDTLS_ERR_SSL_WANT_READ) {
            LOGMSG("tuple %p want read\n", tuple);
            int is_local_server = 0;
            trans_data(tuple, context, is_local_server);
            is_local_server = 1;
            if (trans_data(tuple, context, is_local_server)) return -1;
            return 0;
        } else if (ret == MBEDTLS_ERR_SSL_WANT_WRITE) {
            LOGMSG("tuple %p want write\n", tuple);
            int is_local_server = 0;
            trans_data(tuple, context, is_local_server);
            is_local_server = 1;
            if (trans_data(tuple, context, is_local_server)) return -1;
            return 0;
        } else if (ret <= 0) {
            switch (ret) {
                case MBEDTLS_ERR_SSL_PEER_CLOSE_NOTIFY:
                    LOGERR("tuple %p connection was closed gracefully\n", tuple);
                    context->flag |= FLAG_LOCAL_CLIENT_CLOSE_OK;
                    if (context->flag & FLAG_LOCAL_SERVER_HANDSHAKE_OK && (context->flag & FLAG_LOCAL_SERVER_CLOSE_OK) == 0) {
                        mbedtls_ssl_close_notify(&context->ssl[!is_local_server]);
                        context->flag |= FLAG_NOTIFY_CLIENT_CLOSE;
                    }   
                    break;
                case MBEDTLS_ERR_NET_CONN_RESET:
                    LOGMSG("tuple %p connection was reset by peer\n", tuple);
                    break;
                default:
                    LOGFEW("tuple %p mbedtls_ssl_read returned -0x%x %s\n", tuple, -ret, ssl_err_str(ret));
                    break;
            }   
        } else {
            //tuple->flag |= TUPLE_CT_FLAG_SSL_PROCESS;
            //packet_process_tcp_reply(tuple, buf, ret);
            //tuple->flag &= ~TUPLE_CT_FLAG_SSL_PROCESS;
            int len = ret;
            //LOGERR("got plant data\n"); DUMP(buf, ret);
            //LOGERR("%d %.*s\n", len, len, buf);
            //LOGERR("tuple %p get response %d:\n", tuple, len);
            context->send_len[!is_local_server] += len;
            //LOGERR("tuple %p now response %lu\n", tuple, context->send_to_client_len);
            int is_local_server = 0;
            trans_data(tuple, context, is_local_server);
            is_local_server = 1;
            if (trans_data(tuple, context, is_local_server)) return -1;
        }
    }
    return 0;
}

static inline void clear_sended_packets(SSL_CONTEXT *context, unsigned ack_seq, int is_local_server)
{
    struct rte_mbuf *m;
    //LOGERR("GET_PACKINF(context->sended_to_client_packets)->seq %u   ntohl(tcph->ack_seq) %u\n", GET_PACKINF(context->sended_to_client_packets)->seq, ntohl(tcph->ack_seq));
    assert(context->sended_packet_last[!is_local_server] ? GET_PACKINF(context->sended_packet_last[!is_local_server])->sended_packet_next == NULL : 1);
    while (context->sended_packets[!is_local_server]    && GET_PACKINF(context->sended_packets[!is_local_server])->seq < ack_seq) {
        m = context->sended_packets[!is_local_server];
        context->sended_packets[!is_local_server] = GET_PACKINF(m)->sended_packet_next;
        context->sended_unack_len[!is_local_server] -= GET_PACKINF(m)->app_data_len;
        context->sended_packet_count[!is_local_server]--;
        rte_pktmbuf_free(m);
        fake_pkg_use--;
        LOGERR("fake_pkg_use %d unsend_packet_count %d %d sended_packet_count %d %d\n",
                fake_pkg_use, context->unsend_packet_count[1], context->unsend_packet_count[0], context->sended_packet_count[1], context->sended_packet_count[0]);
        //debug
        assert(context->sended_packet_count[!is_local_server] >= 0);
        assert(fake_pkg_use == context->unsend_packet_count[1] + context->unsend_packet_count[0] + context->sended_packet_count[1] + context->sended_packet_count[0]);
    }
    if (context->sended_packets[!is_local_server] == NULL) context->sended_packet_last[!is_local_server] = NULL;
    //resend_data(tuple, 1);
}

mbedtls_ssl_config *local_ssl_server_conf;
extern mbedtls_ssl_config *local_ssl_client_conf;
extern mbedtls_timing_delay_context *local_server_timer;
/*
 * return 0 ok
 * return 1 cache
 * return x error
 */
static inline int __deal_packet_recved(tuple_t *tuple, struct rte_mbuf *m, int is_local_server)
{
    SSL_CONTEXT *context = tuple->ssl_context;
    struct tcphdr *tcph = GET_TCPH(m);
    uint8_t *app_data = (uint8_t *)tcph + tcph->doff*4;

    if (context->sended_packets[!is_local_server]) clear_sended_packets(context, ntohl(tcph->ack_seq), is_local_server);
    if (APP_DATA_LEN(m) == 0) LOGERR("%s lai cui le, recv_from_server_len %lu send_to_client_len %lu unsend_to_client_packet_count %d recv_from_server_packets %p\n",
                is_local_server ? "client" : "server",
                context->recv_len[!is_local_server],
                context->send_len[!is_local_server],
                context->unsend_packet_count[!is_local_server],
                context->recved_packets[!is_local_server]);
    if (context->unsend_packets[is_local_server]) send_cache_data(tuple, is_local_server);

    if (context->recv_len[is_local_server] + APP_DATA_LEN(m) > sizeof(context->recv_buf[is_local_server])) return 1;
    context->last_deal_pkg_seq_add_len[is_local_server] = ntohl(tcph->seq) + APP_DATA_LEN(m);

    tuple->flag |= TUPLE_CT_FLAG_SSL_STOP_SEND;
    if ((tuple->flag & TUPLE_CT_FLAG_SSL_INIT) == 0) {
        tuple->flag |=TUPLE_CT_FLAG_SSL_INIT;
        int ret;
        {
            mbedtls_ssl_init(&context->ssl[1]);
            context->flag |= LOCAL_SERVER_SSL_INITED;
            if((ret = mbedtls_ssl_setup(&context->ssl[1], local_ssl_server_conf)) != 0)
            { mbedtls_printf(" failed\n  ! mbedtls_ssl_setup returned -0x%x\n\n", -ret); exit(1); }
            mbedtls_ssl_set_bio(&context->ssl[1], tuple, bio_send_to_client, bio_recv_from_client, NULL);
#if defined(MBEDTLS_TIMING_C)
            mbedtls_ssl_set_timer_cb(&context->ssl[1], local_server_timer, mbedtls_timing_set_delay, mbedtls_timing_get_delay);
#endif
        }
        {
            mbedtls_ssl_init( &context->ssl[0] );
            context->flag |= LOCAL_CLIENT_SSL_INITED;
            ret = mbedtls_ssl_setup( &context->ssl[0], local_ssl_client_conf);
            if (ret != 0) { mbedtls_printf( " failed\n  ! mbedtls_ssl_setup returned -0x%x\n\n", -ret ); exit(1); }
            //ret = mbedtls_ssl_set_hostname( &context->local_client_ssl, server_addr );
            //if (ret != 0) { mbedtls_printf( " failed\n  ! mbedtls_ssl_set_hostname returned %d\n\n", ret ); exit(1); }

#if defined(MBEDTLS_KEY_EXCHANGE_ECJPAKE_ENABLED)
            if( ecjpake_pw != DFL_ECJPAKE_PW ) {
                if( ( ret = mbedtls_ssl_set_hs_ecjpake_password( &ssl, (const unsigned char *) ecjpake_pw, strlen( ecjpake_pw ) ) ) != 0 ) {
                    mbedtls_printf( " failed\n  ! mbedtls_ssl_set_hs_ecjpake_password returned %d\n\n", ret );
                    exit(1);
                }
            }
#endif
            mbedtls_ssl_set_bio( &context->ssl[0], tuple, bio_send_to_server, bio_recv_from_server, NULL );
#if defined(MBEDTLS_TIMING_C)
            mbedtls_ssl_set_timer_cb(&context->ssl[0], local_server_timer, mbedtls_timing_set_delay, mbedtls_timing_get_delay);
#endif
        }
    }
    if (APP_DATA_LEN(m) <= 0) {
        if ((is_local_server && (context->flag & FLAG_LOCAL_CLIENT_HANDSHAKE_OK)) || (is_local_server == 0 && (context->flag & FLAG_LOCAL_SERVER_HANDSHAKE_OK))) {
            if (trans_data(tuple, context, is_local_server)) return -1;
        }
        if (trans_data(tuple, context, !is_local_server)) return -1;
    }
    memcpy(context->recv_buf[is_local_server] + context->recv_len[is_local_server], app_data, APP_DATA_LEN(m));
    context->recv_len[is_local_server] += APP_DATA_LEN(m);
    LOGERR("context->recv_from_client_len %lu\n", context->recv_len[is_local_server]);

    if ((context->flag & FLAG_LOCAL_CLIENT_HANDSHAKE_OK) == 0) {
        handshack(tuple, 0);
    }
    if ((context->flag & FLAG_LOCAL_SERVER_HANDSHAKE_OK) == 0) {
        handshack(tuple, 1);
        return 0;
    }
    trans_data(tuple, context, is_local_server);
    if (deal_recved_data(tuple, is_local_server)) return -1;
    trans_data(tuple, context, is_local_server);
    if (trans_data(tuple, context, !is_local_server)) return -1;
    return 0;
} 

static inline int deal_packet_recved(tuple_t *tuple, int is_local_server) 
{
    int deal_count = 0;
    SSL_CONTEXT *context = tuple->ssl_context;
    while (context->recved_packets[is_local_server]) {
        int ret = __deal_packet_recved(tuple, context->recved_packets[is_local_server], is_local_server);
        if (ret == 0) {
            deal_count++;
            struct rte_mbuf *m = context->recved_packets[is_local_server];
            context->recved_packets[is_local_server] = GET_PACKINF(m)->recv_packet_next;
            if (context->recved_packet_last[is_local_server] == m) context->recved_packet_last[is_local_server] = NULL;
            rte_pktmbuf_free(m);
            continue;
        }
        if (ret == 1) {
            if (deal_count) send_to(tuple, (const uint8_t *)"", (size_t)0, 1, 0, 0, is_local_server); //do ack
            return 1;
        }
        return -1;
    }
    if (deal_count) send_to(tuple, (const uint8_t *)"", (size_t)0, 1, 0, 0, is_local_server); //do ack
    return 0;
}
#define deal_packet_recved(a,b) ({ LOGERR("calling deal_packet_recved %d\n", b); deal_packet_recved(a,b); })

static inline int packet_process_ssl(tuple_t *tuple, struct rte_mbuf *m, int is_local_server)
{
    SSL_CONTEXT *context = tuple->ssl_context;
    GET_PACKINF(m)->recv_packet_next = NULL;
    if (context->recved_packets[is_local_server] == NULL) {
        context->recved_packets[is_local_server] = m;
        context->recved_packet_last[is_local_server] = m;
    } else {
        GET_PACKINF(context->recved_packet_last[is_local_server])->recv_packet_next = m;
        context->recved_packet_last[is_local_server] = m;
    }
    switch (deal_packet_recved(tuple, is_local_server)) {
        case 1: return 1;
        case -1: return -1;
                 //default: //must be 0, no error
    }
    switch (deal_packet_recved(tuple, !is_local_server)) {
        case 1: return 1;
        case -1: return -1;
                 //default: //must be 0, no error
    }
    if (tuple->flag & TUPLE_CT_FLAG_CLIENT_FIN) mbedtls_ssl_close_notify(&context->ssl[!is_local_server]);
    return 0;
}
#define packet_process_ssl(a,b,c) ({ LOGERR("calling packet_process_ssl %d\n", c); packet_process_ssl(a,b,c); })

static inline int packet_process_tcp(struct rte_mbuf *m, int queue)
{
    int is_ipv6 = IS_IPV6(m);
    uint32_t srcip[4];
    uint32_t dstip[4];
    struct tcphdr *tcph = GET_TCPH(m);
    static int n = 0; n++; LOGERR("===================================================================================== packet %dth port %d %u --> %u ack %u seq %u app_len %d %s\n", n, m->port, ntohs(tcph->source), ntohs(tcph->dest)
            , ntohl(tcph->ack_seq)
            , ntohl(tcph->seq)
            , APP_DATA_LEN(m)
            , tcph->syn ? "[SYN]" : ""
            );
    if (is_ipv6 == 0) {
        struct iphdr *iph = GET_IPH(m);
        memset(srcip, 0, 12);
        memset(dstip, 0, 12);
        srcip[3] = iph->saddr;
        dstip[3] = iph->daddr;
    } else         {
        struct ip6_hdr *ip6h = GET_IP6H(m);
        memcpy(srcip, &ip6h->ip6_src, 16);
        memcpy(dstip, &ip6h->ip6_dst, 16);
    }
    uint8_t *app_data = (uint8_t *)tcph + tcph->doff*4;
    struct rte_mbuf *next = NULL;

    tuple_t *tuple = find_tuple(m, queue, 6);
    if (tuple) {
        if (tcph->rst && tuple->wait_seq[PACK_DIR(m)]+1 == tcph->seq) goto send; /* guanju said such packet should not abort connections */
        switch (tcp_sort(tuple, m)) {
            case 0: //return 0 ok
                break;
            case 1: //return 1 cache
                goto send;
            case 2: //return 2 full
                goto send;
            case 3: //return 3 same
                goto send;
            case 4: //return 4 fail alloc
                goto send;
            case 100: //return 100 old
                goto send;
            default:
                LOGERR("error\n");
                goto send;
                //if (APP_DATA_LEN(m) == 0) goto loop;
                //if (tuple->flag & TUPLE_CT_FLAG_FILTER_HIT) { rte_pktmbuf_free(m); return; }
                //return 0 ok
                //release_tuple(queue, tuple); tuple = NULL; goto send;
        }
    }
loop:
    next        = GET_PACKINF(m)->tcp_sort_next;
    GET_PACKINF(m)->tcp_sort_next     = NULL;
    tcph        = GET_TCPH(m);
    app_data    = (uint8_t *)tcph + tcph->doff*4;
    if (app_data + APP_DATA_LEN(m) > ((unsigned char *)m->buf_addr) + m->data_off + m->data_len) {
        LOGERR("ERR: app_data + app_data_len > ((unsigned char *)m->buf_addr) + m->data_off + m->data_len\n");
        goto send;
    }
    if (tuple == NULL && tcph->syn && tcph->ack) {
        tuple = create_tuple(m, queue, 6);
        goto send;
    }
    if (PACK_DIR(m) == 0) {
        //LOGERR("one orginal packet\n");
        if (tuple) {
            if (tcph->fin) { tuple->flag |= TUPLE_CT_FLAG_CLIENT_FIN; /*LOGERR("********************client say fin******************\n"); */ }
            /*
            if (tuple->flag & TUPLE_CT_FLAG_SERVER_FIN) {
                if (ntohl(tcph->ack_seq) == tuple->wait_seq[1]) {
                    tuple->flag |= TUPLE_CT_FLAG_SERVER_FIN_ACKED;
                    LOGERR("set TUPLE_CT_FLAG_SERVER_FIN_ACKED\n");
                }
            }
            */
            if (APP_DATA_LEN(m) > 0) tuple->packet_count++;
            switch (tuple->service) {
                case SERVICE_TYPE_HTTP:
                    break;
                case SERVICE_TYPE_SSL:
                    if (packet_process_ssl(tuple, m, 1) < 0) release_tuple(queue, tuple); tuple = NULL; goto end;
                    goto end;
                default:
                    if (tuple->packet_count == 1) {
                        if (tuple->dstport == htons(443)) {
                            tuple->ssl_context = ssl_context_get();
                            if (tuple->ssl_context) {
                                tuple->service = SERVICE_TYPE_SSL;
                                tuple->ssl_context->last_deal_pkg_seq_add_len[0] = ntohl(tcph->ack_seq);
                                tuple->ssl_context->last_deal_pkg_seq_add_len[1] = ntohl(tcph->seq) + APP_DATA_LEN(m);
                                if (packet_process_ssl(tuple, m, 1) < 0) release_tuple(queue, tuple); tuple = NULL; goto end;
                            }
                        }
                    }
                    break;
	    }
            if (tcph->rst) { 
                //LOGERR("original give me a rst, what!!\n");
                release_tuple(queue, tuple); tuple = NULL;
                goto send;
            }
            //if (tuple->flag & TUPLE_CT_FLAG_CLIENT_FIN_ACKED && tuple->flag & TUPLE_CT_FLAG_SERVER_FIN_ACKED) { release_tuple(queue, tuple); tuple = NULL; goto send; }
            if (tuple->flag & TUPLE_CT_FLAG_CLIENT_FIN && tuple->flag & TUPLE_CT_FLAG_SERVER_FIN) { release_tuple(queue, tuple); tuple = NULL; goto send; }
        }
    } else {
        //LOGERR("one reply packet\n");
        if (tuple) {
            if (tcph->fin) { tuple->flag |= TUPLE_CT_FLAG_SERVER_FIN; /*LOGERR("********************server say fin******************\n"); */}
            /*
            if (tuple->flag & TUPLE_CT_FLAG_CLIENT_FIN) {
                LOGERR("one original packet, ack_seq %u wait_seq %u \n", ntohl(tcph->ack_seq), tuple->wait_seq[0]);
                if (ntohl(tcph->ack_seq) == tuple->wait_seq[0]) {
                    tuple->flag |= TUPLE_CT_FLAG_CLIENT_FIN_ACKED;
                    LOGERR("set TUPLE_CT_FLAG_CLIENT_FIN_ACKED\n");
                }
            }
            */
            if (APP_DATA_LEN(m) > 0) tuple->packet_count++;
            switch (tuple->service) {
                case SERVICE_TYPE_SSL:
                    if (packet_process_ssl(tuple, m, 0) < 0) release_tuple(queue, tuple); tuple = NULL; goto end;
                    break;
                default:
                    if (tuple->packet_count == 1) {
                    }
            }
            if (tcph->rst) { release_tuple(queue, tuple); tuple = NULL; goto send; }
            //if (tuple->flag & TUPLE_CT_FLAG_CLIENT_FIN_ACKED && tuple->flag & TUPLE_CT_FLAG_SERVER_FIN_ACKED) { release_tuple(queue, tuple); tuple = NULL; goto send; }
            if (tuple->flag & TUPLE_CT_FLAG_CLIENT_FIN && tuple->flag & TUPLE_CT_FLAG_SERVER_FIN) { release_tuple(queue, tuple); tuple = NULL; goto send; }
        }
    }
    if (tuple && (tuple->flag & TUPLE_CT_FLAG_FILTER_HIT) && !(tuple->flag & TUPLE_CT_FLAG_NOFILTER_HIT)) {
        LOGMSG("server out %u\n", tuple->flag & TUPLE_CT_FLAG_FILTER_HIT);
        rte_pktmbuf_free(m);
        goto end;
    }
send:
    if (tuple && tuple->flag & TUPLE_CT_FLAG_SSL_STOP_SEND) { rte_pktmbuf_free(m); goto end; }
    const uint8_t dst_port = !m->port;
    LOGERR("just forward\n");
    l2fwd_simple_forward(m, m->port);
    packet_process_thread_infos[queue].tx++;
    packet_process_thread_infos[queue].tx_byte += m->data_len;
    port_infos[dst_port].tx++;
    port_infos[dst_port].tx_byte += m->data_len;


end:
    if (next) {
        m = next;
        goto loop;
    }
    return 0;
}

int init_packinf(struct rte_mbuf *m, unsigned queue)
{
    packet_process_thread_infos[queue].rx++;
    packet_process_thread_infos[queue].rx_byte += m->data_len;
    port_infos[m->port].rx++;
    port_infos[m->port].rx_byte += m->data_len;
    memset(GET_PACKINF(m), 0, sizeof(struct packinf));
    int iphdr_len, ippayload_len;
    const struct iphdr *iph = NULL;
    const struct ip6_hdr *ip6h = NULL;
    const uint8_t *data = ((unsigned char *)m->buf_addr) + m->data_off;
    int data_len   = m->data_len;
    if (unlikely(data_len < 42) || memcmp(data, "\xff\xff\xff\xff\xff\xff", 6) == 0) return -1;
    data += 12;
    data_len -= 12;
    int vlan_count = 0;
    L3_OFFSET(m) = m->data_off + 14;
ether:
    switch (*(const uint16_t *) data) {
        case 0x0081:
            data += 4;
            data_len -= 4;
            L3_OFFSET(m) += 4;
            vlan_count++;
            if (vlan_count >= 4) { LOGMSG("we only support 4 vlan header\n"); return -1; }
            goto ether;
        case 0x0008:
            data += 2;
            data_len -= 2;
            iph = (const struct iphdr *)data;
            if (unlikely(iph->version != 4)) return -1;
            iphdr_len = iph->ihl * 4;
            L4_OFFSET(m) = L3_OFFSET(m) + iphdr_len;
            data += iphdr_len;
            data_len -= iphdr_len;
            ippayload_len = be16toh(iph->tot_len) - iphdr_len;
            if (unlikely(data_len > ippayload_len)) data_len = ippayload_len;
            if (iph->protocol == 6) {
                APP_DATA_LEN(m) = ntohs(iph->tot_len) - iph->ihl*4 - GET_TCPH(m)->doff*4;
            } else if (iph->protocol == 17) {
            } else return -1;
            GET_PACKINF(m)->l4_protocol = iph->protocol;
            break;
        case 0xdd86:
            data += 2;
            data_len -= 2;
            ip6h = (const struct ip6_hdr *)data;
            L4_OFFSET(m) = L3_OFFSET(m) + sizeof(struct ip6_hdr);
            data += sizeof(struct ip6_hdr);
            data_len -= sizeof(struct ip6_hdr);
            ippayload_len = be16toh(ip6h->ip6_plen);
            if (ip6h->ip6_nxt == 6) {
                APP_DATA_LEN(m) = ntohs(ip6h->ip6_plen) - GET_TCPH(m)->doff*4;
            } else if (ip6h->ip6_nxt == 17) {
            } else return -1;
            if (unlikely(data_len > ippayload_len)) data_len = ippayload_len;
            IS_IPV6(m) = 1;
            GET_PACKINF(m)->l4_protocol = ip6h->ip6_nxt;
            break;
        default:
            return -1;
    }
    return 0;
}

/*
 * return -1 m not release
 * other     m is release
 */
int packet_process(struct rte_mbuf *m, int queue);
int packet_process(struct rte_mbuf *m, int queue)
{
    switch (GET_PACKINF(m)->l4_protocol) {
        case 6:
            return packet_process_tcp(m, queue);
    }
    return -1;
}

void mim_ssl_client_init(void);
int mim_ssl_server_init(void);
int mim2_init(void);
int mim2_init(void)
{
	g_time = time(NULL);
	mim_ssl_client_init();
	mim_ssl_server_init();
    return 0;
}
